package com.jsean.core.consumer;

import com.jsean.core.repository.H2OffsetRepository;
import com.jsean.core.repository.OffsetRepository;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class OffsetStore {
    private static final Map<String, Long> offsets = new ConcurrentHashMap<>();
    private static final OffsetRepository repository = new H2OffsetRepository();
    private static final ScheduledExecutorService scheduler = 
        Executors.newSingleThreadScheduledExecutor();

    static {
        offsets.putAll(repository.loadAll());
        scheduler.scheduleAtFixedRate(() -> repository.persistAll(offsets), 5, 5, TimeUnit.SECONDS);
    }

    public static long getOffset(String consumerId) {
        return offsets.getOrDefault(consumerId, 0L);
    }

    public static void setOffset(String consumerId, long offset) {
        offsets.put(consumerId, offset);
    }
}